package org.apache.flink.runtime.rest.handler.job;

import java.util.Map;
import java.util.concurrent.CompletableFuture;

import javax.annotation.Nonnull;

import org.apache.flink.api.common.ConsumerID;
import org.apache.flink.api.common.ResourceStatus;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.messages.webmonitor.ResourceStatusInfo;
import org.apache.flink.runtime.resourcemanager.ResourceManagerGateway;
import org.apache.flink.runtime.resourcemanager.ResourceOverview;
import org.apache.flink.runtime.rest.handler.HandlerRequest;
import org.apache.flink.runtime.rest.handler.RestHandlerException;
import org.apache.flink.runtime.rest.handler.resourcemanager.AbstractResourceManagerHandler;
import org.apache.flink.runtime.rest.messages.ConsumerIDPathParameter;
import org.apache.flink.runtime.rest.messages.ConsumerMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
import org.apache.flink.runtime.rest.messages.MessageHeaders;
import org.apache.flink.runtime.rpc.retriever.GatewayRetriever;
import org.apache.flink.runtime.webmonitor.RestfulGateway;

/** Handler which serves detailed TaskManager information. */
public class ResourceStatusHandler
        extends AbstractResourceManagerHandler<
                RestfulGateway,
                EmptyRequestBody,
                ResourceStatusInfo,
                ConsumerMessageParameters> {

    public ResourceStatusHandler(
            GatewayRetriever<? extends RestfulGateway> leaderRetriever,
            Time timeout,
            Map<String, String> responseHeaders,
            MessageHeaders<EmptyRequestBody, ResourceStatusInfo, ConsumerMessageParameters> messageHeaders,
            GatewayRetriever<ResourceManagerGateway> resourceManagerGatewayRetriever) {
        super(
                leaderRetriever,
                timeout,
                responseHeaders,
                messageHeaders,
                resourceManagerGatewayRetriever);

    }

    @Override
    protected CompletableFuture<ResourceStatusInfo> handleRequest(
            @Nonnull HandlerRequest<EmptyRequestBody> request,
            @Nonnull ResourceManagerGateway gateway)
            throws RestHandlerException {

        CompletableFuture<ResourceOverview> futrue =
                	gateway.requestResourceOverview(timeout);

        return futrue
                .thenApply(
                		resourceOverview ->  ResourceStatus.from(resourceOverview.getNumberRegisteredSlots(), 
                				resourceOverview.getNumberRegisteredSlots() - resourceOverview.getNumberFreeSlots()))
                .thenApply(ResourceStatusInfo::new);
    }

    
}
